/**
 * Copyright (c) 2013-2020 Contributors to the Eclipse Foundation
 *
 * <p> See the NOTICE file distributed with this work for additional information regarding copyright
 * ownership. All rights reserved. This program and the accompanying materials are made available
 * under the terms of the Apache License, Version 2.0 which accompanies this distribution and is
 * available at http://www.apache.org/licenses/LICENSE-2.0.txt
 */
package org.locationtech.geowave.datastore.accumulo;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.SortedMap;
import java.util.TreeMap;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Value;
import org.locationtech.geowave.core.store.entities.GeoWaveKey;
import org.locationtech.geowave.core.store.entities.GeoWaveKeyImpl;
import org.locationtech.geowave.core.store.entities.GeoWaveRow;
import org.locationtech.geowave.core.store.entities.GeoWaveValue;
import org.locationtech.geowave.core.store.entities.GeoWaveValueImpl;

public class AccumuloRow implements GeoWaveRow {
  private final GeoWaveKey key;
  private GeoWaveValue[] fieldValues;

  private static class LatestFirstComparator implements Comparator<Long>, Serializable {
    /**
     *
     */
    private static final long serialVersionUID = 1L;

    @Override
    public int compare(final Long ts1, final Long ts2) {
      return ts2.compareTo(ts1);
    }
  }

  public AccumuloRow(
      final byte[] rowBytes,
      final int partitionKeyLength,
      final List<Map<Key, Value>> fieldValueMapList,
      final boolean sortByTime) {
    // TODO: GEOWAVE-1018 - can we do something more clever that lazily
    // parses only whats required by the getter (and caches anything else
    // that is parsed)?
    key = new GeoWaveKeyImpl(rowBytes, partitionKeyLength);

    if (sortByTime) {
      setTimeSortedFieldValues(fieldValueMapList);
    } else {
      setFieldValues(fieldValueMapList);
    }
  }

  private void setFieldValues(final List<Map<Key, Value>> fieldValueMapList) {
    final List<GeoWaveValue> fieldValueList = new ArrayList();

    for (final Map<Key, Value> kvMap : fieldValueMapList) {
      for (final Entry<Key, Value> kv : kvMap.entrySet()) {
        fieldValueList.add(
            new GeoWaveValueImpl(
                kv.getKey().getColumnQualifier().getBytes(),
                kv.getKey().getColumnVisibility().getBytes(),
                kv.getValue().get()));
      }
    }

    fieldValues = new GeoWaveValue[fieldValueList.size()];
    int i = 0;

    for (final GeoWaveValue gwValue : fieldValueList) {
      fieldValues[i++] = gwValue;
    }
  }

  private void setTimeSortedFieldValues(final List<Map<Key, Value>> fieldValueMapList) {
    final SortedMap<Long, GeoWaveValue> fieldValueSortedMap =
        new TreeMap(new LatestFirstComparator());

    for (final Map<Key, Value> kvMap : fieldValueMapList) {
      for (final Entry<Key, Value> kv : kvMap.entrySet()) {
        fieldValueSortedMap.put(
            kv.getKey().getTimestamp(),
            new GeoWaveValueImpl(
                kv.getKey().getColumnQualifier().getBytes(),
                kv.getKey().getColumnVisibility().getBytes(),
                kv.getValue().get()));
      }
    }

    final Iterator it = fieldValueSortedMap.entrySet().iterator();

    fieldValues = new GeoWaveValue[fieldValueSortedMap.size()];
    int i = 0;

    while (it.hasNext()) {
      final Map.Entry entry = (Map.Entry) it.next();
      final GeoWaveValue gwValue = (GeoWaveValue) entry.getValue();
      fieldValues[i++] = gwValue;
    }
  }

  @Override
  public byte[] getDataId() {
    return key.getDataId();
  }

  @Override
  public short getAdapterId() {
    return key.getAdapterId();
  }

  @Override
  public byte[] getSortKey() {
    return key.getSortKey();
  }

  @Override
  public byte[] getPartitionKey() {
    return key.getPartitionKey();
  }

  @Override
  public int getNumberOfDuplicates() {
    return key.getNumberOfDuplicates();
  }

  @Override
  public GeoWaveValue[] getFieldValues() {
    return fieldValues;
  }
}
